package com.czk.rocketmq.delay;

import com.czk.rocketmq.order.OrderStep;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * 发送延迟消息
 * @Author:ChenZhangKun
 * @Date: 2020/9/5 13:33
 */
@SuppressWarnings("all")
public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        // 1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        // 2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.25.133:9876;192.168.25.134:9876");
        // 3.启动producer
        producer.start();
        // 拿到orders,构建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();
        // 发送消息
        for (int i = 0; i < orderSteps.size(); i++) {
            // 构造message
            Message message=new Message("delayTopic","Order","i"+i,(orderSteps.get(i)+"").getBytes());
            // 设置延迟时间
            message.setDelayTimeLevel(2);
            /**
             * 参数一：消息对象
             * 参数二：消息队列选择器
             * 参数三：业务标识参数
             */
            SendResult result= producer.send(message, new MessageQueueSelector() {
                /**
                 *
                 * @param list 队列集合
                 * @param message 消息对象
                 * @param o 业务标识参数
                 * @return 选择的消息队列
                 */
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    // 强转成订单id
                    long orderId=(long)o;
                    // 订单id取模->只是针对目前的简单业务而言
                    long index=orderId%list.size();
                    return list.get((int) index);
                }
            },orderSteps.get(i).getOrderId());
            System.out.println("发送结果是"+result);
        }
        // 关闭生产者
        producer.shutdown();
    }
}
